﻿/*
    Copyright (c) 2012, Matías López
    All rights reserved.

    Redistribution and use in source and binary forms, with or without modification, are permitted 
    provided that the following conditions are met:
        *   Redistributions of source code must retain the above copyright notice, this list of 
            conditions and the following disclaimer.
        *   Redistributions in binary form must reproduce the above copyright notice, this list of 
            conditions and the following disclaimer in the documentation and/or other materials 
            provided with the distribution.
        *   Neither the name of Matías López nor the names of its contributors may be used to 
            endorse or promote products derived from this software without specific prior written 
            permission.
    
    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR 
    IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY 
    AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR 
    CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
    CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 
    SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 
    THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR 
    OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
    POSSIBILITY OF SUCH DAMAGE.
 */

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.StorageClient;

namespace mana.Data.Azure.Queues
{
    /// <summary>
    /// Exposes main queue operations using an Azure Queue.
    /// </summary>
    /// <typeparam name="T">Object type that the queue will handles.</typeparam>
    public abstract class GenericQueue<T>
    {
        private static class Validations
        {
            public static void Name(string value)
            {
                if (string.IsNullOrWhiteSpace(value))
                    throw new Exception("Queue name not specified.");
            }
        }

        /// <summary>
        /// Raised when a poison message is found. (The poison message will be automatically deleted from the queue).
        /// </summary>
        public event EventHandler<GenericQueuePoisonMessageEventArgs> PoisonMessageFound;

        private CloudStorageAccount _account;
        private string _name;

        private CloudQueue _cloudQueue;
        private GenericQueueProcessor<T> _processor;

        /// <summary>
        /// Initializes a GenericQueue instance.
        /// </summary>
        /// <param name="account">Targeted Cloud Storage Account.</param>
        /// <param name="name">Azure Queue name.</param>
        public GenericQueue(CloudStorageAccount account, string name)
        {
            Validations.Name(name);

            _account = account;
            _name = name;

            _cloudQueue = account.CreateCloudQueueClient().GetQueueReference(name);
            _cloudQueue.CreateIfNotExist();

            this.PoisonMessageFound += new EventHandler<GenericQueuePoisonMessageEventArgs>(GenericQueue_PoisonMessageFound);
        }

        #region Properties
        
        /// <summary>
        /// Gets the Azure Queue name.
        /// </summary>
        public string QueueName
        {
            get { return _name; }
        }

        /// <summary>
        /// Gets a reference to the internal Cloud Queue.
        /// </summary>
        public CloudQueue AzureQueue
        {
            get { return _cloudQueue; }
        }

        /// <summary>
        /// Gets the targeted Cloud Storage Account.
        /// </summary>
        public CloudStorageAccount Account
        {
            get { return _account; }
        }

        /// <summary>
        /// Azure Queue processor.
        /// </summary>
        public GenericQueueProcessor<T> Processor
        {
            get
            {
                if (_processor == null)
                    _processor = new GenericQueueProcessor<T>(this);

                return _processor;
            }
        }

        #endregion Properties

        #region Methods
        
        /// <summary>
        /// Enqueues an object into the queue.
        /// </summary>
        /// <param name="obj">Object to enqueue.</param>
        public void Enqueue(T obj)
        {
            string message = this.GetMessageFromObj(obj);
            _cloudQueue.AddMessage(new CloudQueueMessage(message));
        }

        /// <summary>
        /// Enqueues an object into the queue with the specified time to live.
        /// </summary>
        /// <param name="obj">Object to enqueue.</param>
        /// <param name="timeToLive">Time for the object to live in the queue.</param>
        public void Enqueue(T obj, TimeSpan timeToLive)
        {
            string message = this.GetMessageFromObj(obj);
            _cloudQueue.AddMessage(new CloudQueueMessage(message), timeToLive);
        }

        /// <summary>
        /// Enqueues a list of objects into the queue.
        /// </summary>
        /// <param name="objs">List of object to enqueue.</param>
        public void Enqueue(IEnumerable<T> objs)
        {
            foreach (T obj in objs)
            {
                this.Enqueue(obj);
            }
        }

        /// <summary>
        /// Enqueues a list of objects into the queue with the specified time to live.
        /// </summary>
        /// <param name="objs">List of object to enqueue.</param>
        /// <param name="timeToLive">Time for every object to live in the queue.</param>
        public void Enqueue(IEnumerable<T> objs, TimeSpan timeToLive)
        {
            foreach (T obj in objs)
            {
                this.Enqueue(obj, timeToLive);
            }
        }

        /// <summary>
        /// Dequeues an object from the queue. This is the first part of the "Dequeue" process. MUST be followed by "EndDequeue" method to complete the process after processing the dequeued object.
        /// </summary>
        /// <param name="receipt">Dequeue receipt needed to end the "Dequeue" process.</param>
        /// <returns>Returns the dequeued object.</returns>
        public T BeginDequeue(out DequeueReceipt receipt)
        {
            return BeginDequeue(TimeSpan.FromSeconds(30), out receipt);
        }

        /// <summary>
        /// Dequeues an object from the queue. This is the first part of the "Dequeue" process. MUST be followed by "EndDequeue" method to complete the process after processing the dequeued object.
        /// </summary>
        /// <param name="visibilityTimeout">Visibility timeout for dequeueing the object.</param>
        /// <param name="receipt">Dequeue receipt needed to end the "Dequeue" process.</param>
        /// <returns>Returns the dequeued object.</returns>
        public T BeginDequeue(TimeSpan visibilityTimeout, out DequeueReceipt receipt)
        {
            receipt = new DequeueReceipt() { MessageId = null, PopReceipt = null };

            T result = default(T);

            CloudQueueMessage message;
            bool isPoisonMessage;
            do
            {
                message = _cloudQueue.GetMessage(visibilityTimeout);
                if (message == null)
                    return result;

                isPoisonMessage = false;

                //  If message was dequeued MORE THAN 4 TIMES is a Poison Message
                if (message.DequeueCount > 4)
                {
                    isPoisonMessage = true;
                    this.RaisePoisonMessageFoundEvent(message);
                    continue;
                }

                if (IsPoisonMessage(message))
                {
                    isPoisonMessage = true;
                    this.RaisePoisonMessageFoundEvent(message);
                    continue;
                }

                result = this.GetObjFromMessage(message);
            }
            while (isPoisonMessage);

            receipt.MessageId = message.Id;
            receipt.PopReceipt = message.PopReceipt;
            
            return result;
        }

        /// <summary>
        /// Deletes the dequeued object from the queue. This is the second part of the "Dequeue" process. MUST follow "BeginDequeue" method to complete the process.
        /// </summary>
        /// <param name="receipt">The dequeue receipt obtained from the "BeginDequeue" method.</param>
        public void EndDequeue(DequeueReceipt receipt)
        {
            _cloudQueue.DeleteMessage(receipt.MessageId, receipt.PopReceipt);
        }

        /// <summary>
        /// Gets the message string to enqueue from the specified object.
        /// </summary>
        /// <param name="obj">Object to enqueue.</param>
        /// <returns></returns>
        protected abstract string GetMessageFromObj(T obj);

        /// <summary>
        /// Gets the object from the message string.
        /// </summary>
        /// <param name="message">Queue message.</param>
        /// <returns></returns>
        protected abstract T GetObjFromMessage(CloudQueueMessage message);

        /// <summary>
        /// Indicates whether the message is considered a poison message or not. (The queue will automatically consider a message with a "DequeueCount" greater than 4 as poisoned).
        /// </summary>
        /// <param name="message">Queue message.</param>
        /// <returns></returns>
        protected abstract bool IsPoisonMessage(CloudQueueMessage message);
        
        protected virtual void OnPoisonMessageFound(GenericQueuePoisonMessageEventArgs e)
        {
            if (this.PoisonMessageFound != null)
                this.PoisonMessageFound(this, e);
        }

        private void RaisePoisonMessageFoundEvent(CloudQueueMessage message)
        {
            this.OnPoisonMessageFound(new GenericQueuePoisonMessageEventArgs() { Message = message });
        }

        #endregion Methods

        #region Event Handlers
        
        protected void GenericQueue_PoisonMessageFound(object sender, GenericQueuePoisonMessageEventArgs e)
        {
            try
            {
                _cloudQueue.DeleteMessage(e.Message);
            }
            catch
            {
            }
        }

        #endregion Event Handlers
    }

    public class GenericQueuePoisonMessageEventArgs : EventArgs
    {
        public CloudQueueMessage Message { get; set; }
    }

    /// <summary>
    /// Queue dequeue receipt.
    /// </summary>
    public struct DequeueReceipt
    {
        public string MessageId;
        public string PopReceipt;
    }
}
